package com.duowan.realtime.uniq.algorithm.bloomfilter;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.OutputStream;
import java.io.Serializable;
import java.nio.charset.Charset;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.BitSet;
import java.util.Collection;
import java.util.Random;

import org.apache.commons.io.FileUtils;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang.builder.ToStringBuilder;

import com.duowan.realtime.hash.AbstractCustomHashFunction;
import com.duowan.realtime.hash.RandomNumberHashFunction;
import com.duowan.realtime.hash.SecureRandomNumberHashFunction;
import com.duowan.realtime.hash.SimpleLcgHashFunction;


public class BloomFilter<T> implements Cloneable, Serializable {
	private static final long serialVersionUID = -6910506929951815483L;
	
	protected BitSet bloom;
	protected transient MessageDigest hashFunction;
	protected String hashFunctionName;
	protected int k;
	protected int m;
	private transient Charset defaultCharset;
	protected String defaultCharsetName = "UTF-8";
	protected HashMethod hashMethod = HashMethod.Cryptographic;
	private AbstractCustomHashFunction customHashFunction;
	protected final static int seed32 = 89478583;
	private int numberOfAddedElements; // number of elements actually added to the Bloom filter


	private boolean changed = false;
	/**
	 * n个元素，positive rate为p，M的最优值
	 */
	public static int optimalM(double n, double p) {
		return (int) Math.ceil(-1 * n * Math.log(p) / Math.pow(Math.log(2), 2));
	}

	/**
	 * k = ln(2)* m/n 时出错的概率
	 */
	public static int optimalK(double n, int m) {
		return (int) Math.ceil(Math.log(2) * m / n);
	}

	/**
	 * 
	 * @param n Expected number of elements inserted in the bloom filter
	 * @param p Tolerable false positive rate
	 */
	public BloomFilter(double n, double p) {
		this(optimalM(n, p), optimalK(n, optimalM(n, p)));
	}

	/**
	 * 
	 * @param m	The size of the bloom filter in bits.
	 * @param k	The number of hash functions to use.
	 */
	public BloomFilter(int m, int k) {
		this(new BitSet(m), m, k);
	}

	public BloomFilter(BitSet bloom, int m, int k) {
		this.m = m;
		this.bloom = bloom;
		this.k = k;
		setCryptographicHashFunction("MD5");
	}

	public BloomFilter(BitSet bloomFilter, int m, int k, HashMethod hashMethod, String hashFunctionName) {
		this.bloom = bloomFilter;
		this.k = k;
		this.m = m;
		setHashMethod(hashMethod);
		setCryptographicHashFunction(hashFunctionName);
	}

	/**
	 * 默认构造函数
	 */
	protected BloomFilter() {
	}

	public void setHashMethod(HashMethod hashMethod) {
		this.hashMethod = hashMethod;
	}

	public void setCryptographicHashFunction(String hashFunctionName) {
		this.hashMethod = HashMethod.Cryptographic;
		this.hashFunctionName = hashFunctionName;
		try {
			hashFunction = MessageDigest.getInstance(hashFunctionName);
		} catch (NoSuchAlgorithmException e) {
			throw new RuntimeException(
					"Unknown hash function provided. Use  MD2, MD5, SHA-1, SHA-256, SHA-384 or SHA-512.");
		}
	}

	public void setCusomHashFunction(AbstractCustomHashFunction chf) {
		this.hashMethod = HashMethod.Custom;
		this.customHashFunction = chf;
	}

	public synchronized boolean add(byte[] value) {
		changed = true;
		for (int position : hash(value)) {
			setBit(position);
		}
		numberOfAddedElements++;
		
		return true;
	}

	public synchronized boolean add(T value) {
		return add(value.toString().getBytes(getDefaultCharset()));
	}

	public void addAll(Collection<T> values) {
		for (T value : values){
			add(value);
		}
	}

	public void clear() {
		bloom.clear();
	}

	public void destory() {
		bloom = null ;
	}
	
	public synchronized boolean contains(byte[] value) {
		for (int position : hash(value))
			if (!getBit(position))
				return false;
		return true;
	}

	public synchronized boolean contains(T value) {
		return contains(value.toString().getBytes(getDefaultCharset()));
	}

	public boolean containsAll(Collection<T> values) {
		for (T value : values){
			if (!contains(value)){
				return false;
			}
		}
		return true;
	}

	protected boolean getBit(int index) {
		return bloom.get(index);
	}

	protected Charset getDefaultCharset() {

		if (this.defaultCharset == null) {
			this.defaultCharset = Charset.forName(defaultCharsetName);
		}

		return this.defaultCharset;
	}

	protected void setBit(int index) {
		setBit(index, true);
	}

	protected void setBit(int index, boolean to) {
		bloom.set(index, to);
		changed = true;
	}

	public BitSet getBitSet() {
		return bloom;
	}

	public int[] hash(String value) {
		return hash(value.getBytes(getDefaultCharset()));
	}

	/**
	 * 选择hash函数
	 */
	protected int[] hash(byte[] value) {
		// Dispatch the chosen hash method
		switch (hashMethod) {
		case RNG:
			return new RandomNumberHashFunction().hash(value, m, k);
		case SecureRNG:
			return new SecureRandomNumberHashFunction().hash(value, m, k);
		case SimpeLCG:
			return new SimpleLcgHashFunction().hash(value, m, k);
		case Custom:
			return customHashFunction.hash(value, m, k);
		default:
			return hashCrypt(value);
		}
	}

	public synchronized boolean union(BloomFilter<T> other) {
		if (compatible(this, other)) {
			bloom.or(other.bloom);
			return true;
		}
		return false;
	}

	public synchronized boolean intersect(BloomFilter<T> other) {
		if (compatible(this, other)) {
			bloom.and(other.bloom);
			return true;
		}
		return false;
	}

	protected boolean compatible(BloomFilter<T> bloomFilter, BloomFilter<T> other) {
		return this.m == other.m && this.k == other.k
				&& this.hashFunctionName.equals(other.hashFunctionName)
				&& this.hashMethod == other.hashMethod;
	}

	public synchronized boolean isEmpty() {
		return bloom.isEmpty();
	}

	public double getFalsePositiveProbability(int n) {
		return Math.pow((1 - Math.exp(-k * (double) n / (double) m)), k);
	}

	public double getBitsPerElement(int n) {
		return m / (double) n;
	}

	public double getBitZeroProbability(int n) {
		return Math.pow(1 - (double) 1 / m, k * n);
	}

	/**
	 * Returns the size of the bloom filter.
	 * 
	 * @return Size of the bloom filter in bits
	 */
	public int size() {
		return m;
	}

	/**
	 * Returns the size of the bloom filter.
	 * 
	 * @return Size of the bloom filter in bits
	 */
	public int getM() {
		return m;
	}

	/**
	 * Returns the number of hash functions.
	 * 
	 * @return number of hash functions used
	 */
	public int getK() {
		return k;
	}

	public String getCryptographicHashFunctionName() {
		return hashFunctionName;
	}

	public MessageDigest getCryptographicHashFunction() {

		if (this.hashFunction == null) {
			setCryptographicHashFunction(hashFunctionName);
		}

		return this.hashFunction;
	}

	/**
	 * cryptographic hash function
	 */
	protected int[] hashCrypt(byte[] value) {
		int[] positions = new int[k];

		int computedHashes = 0;
		// Add salt to the hash deterministically in order to generate different
		// hashes for each round
		// Alternative: use pseudorandom sequence
		Random r = getRandom();
		byte[] digest = new byte[0];
		while (computedHashes < k) {
			// byte[] saltBytes =
			// ByteBuffer.allocate(4).putInt(r.nextInt()).array();
			synchronized (getCryptographicHashFunction()) {
				getCryptographicHashFunction().update(digest);
                digest = getCryptographicHashFunction().digest(value);              
            }
			//getCryptographicHashFunction().update(digest);
			//digest = getCryptographicHashFunction().digest(value);
			BitSet hashed = BitSet.valueOf(digest);

			// Convert the hash to numbers in the range [0,m)
			// Size of the Bloomfilter rounded to the next power of two
			int filterSize = 32 - Integer.numberOfLeadingZeros(m);
			// Computed hash bits
			int hashBits = digest.length * 8;
			// Split the hash value according to the size of the Bloomfilter
			for (int split = 0; split < (hashBits / filterSize) && computedHashes < k; split++) {
				int from = split * filterSize;
				int to = (split + 1) * filterSize;
				BitSet hashSlice = hashed.get(from, to);
				// Bitset to Int
				long[] longHash = hashSlice.toLongArray();
				int intHash = longHash.length > 0 ? (int) longHash[0] : 0;
				// Only use the position if it's in [0,m); Called rejection
				// sampling
				if (intHash < m) {
					positions[computedHashes] = intHash;
					computedHashes++;
				}
			}
		}

		return positions;
	}

	/**
	 * Returns the hash method used to calculate hash values.
	 * 
	 * @return the hash method used to calculate hash values.
	 */
	public HashMethod getHashMethod() {
		return hashMethod;
	}

	protected Random getRandom() {
		return new Random(160598551545387l);
	}

	@Override
	@SuppressWarnings("unchecked")
	public synchronized Object clone() {
		BloomFilter<T> o = null;
		try {
			o = (BloomFilter<T>) super.clone();
		} catch (CloneNotSupportedException e) {
		}
		o.bloom = (BitSet) bloom.clone();
		o.setCryptographicHashFunction(this.hashFunctionName);
		o.k = k;
		o.m = m;
		return o;
	}

	@Override
	public synchronized int hashCode() {
		final int prime = 31;
		int result = 1;
		result = prime * result + ((bloom == null) ? 0 : bloom.hashCode());
		result = prime * result + ((hashFunctionName == null) ? 0 : hashFunctionName.hashCode());
		result = prime * result + k;
		result = prime * result + m;
		return result;
	}

	@Override
	public synchronized boolean equals(Object obj) {
		if (this == obj)
			return true;
		if (obj == null)
			return false;
		if (getClass() != obj.getClass())
			return false;
		BloomFilter other = (BloomFilter) obj;
		if (bloom == null) {
			if (other.bloom != null)
				return false;
		} else if (!bloom.equals(other.bloom))
			return false;
		if (hashFunctionName == null) {
			if (other.hashFunctionName != null)
				return false;
		} else if (!hashFunctionName.equals(other.hashFunctionName))
			return false;
		if (k != other.k)
			return false;
		if (m != other.m)
			return false;
		return true;
	}

	@Override
	public synchronized String toString() {
		StringBuilder sb = new StringBuilder();
		sb.append("Bloom Filter, Parameters ");
		sb.append("m = " + getM() + ", ");
		sb.append("k = " + getK());
		return sb.toString();
	}

	/**
	 * Different types of hash functions that can be used.
	 */
	public static enum HashMethod {
		RNG("RNG"), 
		SecureRNG("SecureRNG"), 
		Cryptographic("Cryptographic"), 
		SimpeLCG("SimpleLCG"), 
		Custom("Custom");

		private String name;

		private HashMethod(String name) {
			this.name = name;
		}

		@Override
		public String toString() {
			return name;
		}
	}
	
	public void writeBitSetToFile(String fileName) throws IOException {
		writeBitSetToFile(fileName, this.getBitSet());
	}

	public static void writeBitSetToFile(String fileName, BitSet bitSet) throws IOException {
		//java 7 的新方法
//		FileUtils.writeByteArrayToFile(new File(fileName), bitSet.toByteArray());
		File file = new File(fileName);
		if (!file.getParentFile().exists()) {
			FileUtils.forceMkdir(file.getParentFile());
		}
		writeBitSetToOutputStream(new FileOutputStream(file, false), bitSet) ;
	}
	
	public void writeBitSetToOutputStream(OutputStream outputStream) throws IOException {
		writeBitSetToOutputStream(outputStream, this.getBitSet());
	}
	public static void writeBitSetToOutputStream(OutputStream outputStream, BitSet bitSet) throws IOException {
		
		ObjectOutputStream objectOutputStream = null ;
		try {
			objectOutputStream = new ObjectOutputStream(outputStream);
			objectOutputStream.writeObject(bitSet);
			objectOutputStream.flush();
		}catch (FileNotFoundException e) {
			throw new RuntimeException(e);
		}  
		finally {
			IOUtils.closeQuietly(objectOutputStream);
		}
	}

	public static BitSet readBitSetFromFile(String fileName) throws IOException{
		return readBitSetFromFile(new File(fileName));
	}
	
	public static BitSet readBitSetFromFile(File file) throws IOException{
//		return BitSet.valueOf(FileUtils.readFileToByteArray(file));
		
		if (!file.exists()) {
			throw new FileNotFoundException(" file " + file + " does not exits. " );
		}
		
		return readBitSetFromInputStream(new FileInputStream(file)) ;
	}
	
	public static BitSet readBitSetFromInputStream(InputStream inputStream ) throws IOException{
		ObjectInputStream objectInputStream = null; 
		try {
			objectInputStream = new ObjectInputStream(inputStream);
			return (BitSet) objectInputStream.readObject();
		} catch (ClassNotFoundException e) {
			throw new RuntimeException(e);
		}finally {
			IOUtils.closeQuietly(objectInputStream);
		}
	}

	public boolean isChange() {
		return changed;
	}
	
	public void cleanChange() {
		changed = false;
	}

	public int getNumberOfAddedElements() {
		return numberOfAddedElements;
	}
}
